﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Xml.Serialization;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using RuTracker.Messenger.Error;

namespace RuTracker.Messenger
{
    /// <summary>
    /// Messenger for Windows Azure bus queue
    /// </summary>
    public class QueueMessenger : IDisposable
    {
        private readonly Queue<object> _messagesToSend;
        private readonly NamespaceManager _namespaceManager;
        private readonly string _queueName;
        private readonly QueueClient _client;

        public QueueMessenger(string connectionString, string queueName, ReceiveMode receiveMode = ReceiveMode.ReceiveAndDelete)
        {
            if (string.IsNullOrEmpty(connectionString))
                throw new ArgumentNullException("connectionString");

            if (string.IsNullOrEmpty(queueName))
                throw new ArgumentNullException("queueName");

            _queueName = queueName;
            _messagesToSend = new Queue<object>();
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
            EnsureQueueExists();
            _client = QueueClient.CreateFromConnectionString(connectionString, _queueName, receiveMode);
        }

        /// <summary>
        /// Receives specific number of messages from queue, if count is less than 0 - all messages will be received
        /// </summary>
        /// <param name="count"></param>
        public QueueMessagesResult<T> Receive<T>(int count = -1)
        {
            int messagesReceived = 0;
            
            List<T> messages = new List<T>();
            List<MessageReadError> errors = new List<MessageReadError>();
            BrokeredMessage message;
            while (count < 0 || (count > 0 && messagesReceived < count) )
            {

                try
                {
                    message = _client.Receive();
                    if (message == null)
                        break;

                    string body = message.GetBody<string>();
                    messages.Add(MessageFromString<T>(body));
                }
                catch (Exception ex)
                {
                    errors.Add(new MessageReadError(ex.ToString()));
                }

                messagesReceived++;
            }

            return new QueueMessagesResult<T>(messages, new MessageErrorCollection(errors));
        }

        /// <summary>
        /// Adds message to queue
        /// </summary>
        public void AddMessage(object message)
        {
            _messagesToSend.Enqueue(message);
        }

        /// <summary>
        /// Adds messages to queue
        /// </summary>
        public void AddMessages(IEnumerable<object> messages)
        {
            foreach (var message in messages)
            {
                AddMessage(message);   
            }
        }

        /// <summary>
        /// Forces messages sending
        /// </summary>
        public void Send()
        {
            EnsureQueueExists();

            for (int i = 0; i < _messagesToSend.Count; i++)
            {
                string body = MessageToString(_messagesToSend.Dequeue());
                BrokeredMessage brokeredMessage = new BrokeredMessage(body);
                brokeredMessage.Properties["SequenceId"] = i;
                _client.Send(brokeredMessage);
            }
        }

        /// <summary>
        /// disposes connection
        /// </summary>
        public void Dispose()
        {
            Send();
            _client.Close();
        }

        public void Purge()
        {
            BrokeredMessage message = _client.Receive();

            while (message != null)
            {
                
                try
                {
                    message = _client.Receive();
                }
                catch
                {
                    //just ignore any errors
                }
            }
        }

        private void EnsureQueueExists()
        {
            if (!_namespaceManager.QueueExists(_queueName))
            {
                QueueDescription description = new QueueDescription(_queueName)
                {
                    EnableBatchedOperations = false
                };

                _namespaceManager.CreateQueue(description);
            }
        }

        private T MessageFromString<T>(string body)
        {
            try
            {
                XmlSerializer serializer = new XmlSerializer(typeof(T));
                using (StringReader SR = new StringReader(body))
                {
                    return (T)serializer.Deserialize(SR);
                }
            }
            catch (Exception ex)
            {
                throw new QueueMessengerException("Message cannot be deserialized properly", ex);
            }
        }

        private string MessageToString(object message)
        {
            try
            {
                XmlSerializer serializer = new XmlSerializer(message.GetType());
                using (StringWriter SR = new StringWriter())
                {
                    serializer.Serialize(SR, message);
                    return SR.ToString();
                }
            }
            catch (Exception ex)
            {
                throw new QueueMessengerException("Message cannot be serialized properly", ex);
            }
        }
    }
}
